package main

import (
	"encoding/json"
	"github.com/Shopify/sarama"
	"log"
)

var broker = []string{"127.0.0.1:9092"}

var (
	KafkaProducer sarama.AsyncProducer
	KafkaClient   sarama.Client
	quit          = make(chan bool)
)

func InitKafka() {

	var err error
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	KafkaClient, err = sarama.NewClient(broker, config)
	if err != nil {
		panic(err.Error())
	}

	KafkaProducer, err = sarama.NewAsyncProducerFromClient(KafkaClient)
	if err != nil {
		panic(err.Error())
	}

	go func(p sarama.AsyncProducer, quit chan bool) {
		for {
			select {
			case suc := <-p.Successes():
				//case  suc := <-p.Successes():
				log.Print("offset: ", suc.Offset, "timestamp: ", suc.Timestamp.String(), "partitions: ", suc.Partition)
			case fail := <-p.Errors():
				log.Printf("AsyncProducer receive err: %v", fail.Err)
			case <-quit:
				log.Printf("AsyncProducer return---------------")
				p.AsyncClose()
				return
			}
		}
	}(KafkaProducer, quit)
}
func ShutDownKafka() {
	if KafkaProducer == nil || KafkaProducer.Input() == nil {
		return
	}
	quit <- true
	if KafkaClient != nil {
		err := KafkaClient.Close()
		if err != nil {
			log.Printf("ShutDownKafka,err:%s", err.Error())
		}
	}
}

func Send(msgBody interface{}, topic string) {
	if KafkaProducer == nil || KafkaProducer.Input() == nil {
		return
	}
	value, err := json.Marshal(msgBody)
	if err != nil {
		log.Print("kafka send msg json.Marshal error", map[string]interface{}{"err": err})
		return
	}
	msg := &sarama.ProducerMessage{
		Topic: topic,
		Value: sarama.ByteEncoder(value),
	}
	//使用通道发送
	KafkaProducer.Input() <- msg
}
